package com.example.kafka;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

public class Customer1 {

    public static void main(String[] args) {
        Properties prop = new Properties();

        // 设置 Kafka 消费者配置
        prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.200.130:9093");
        prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        prop.put(ConsumerConfig.GROUP_ID_CONFIG, "con-1"); // 消费者组
        prop.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest"); // 设置从最新偏移量开始消费
        prop.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); // 禁用自动提交偏移量，手动提交
        prop.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000); // 自动提交间隔时间

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(prop);
        ArrayList<String> topics = new ArrayList<>();
        topics.add("fr");  // 订阅多个主题
        consumer.subscribe(topics);

        try {
            while (true) {
                // 拉取消息
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(10));

                // 遍历每个分区的消息
                for (TopicPartition partition : records.partitions()) {
                    List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);

                    String topic = partition.topic();
                    int partitionId = partition.partition();
                    int size = partitionRecords.size();
                    System.out.printf("--- 获取topic: %s, 分区位置：%d, 消息总数：%d%n", topic, partitionId, size);

                    // 处理每条消息
                    for (ConsumerRecord<String, String> record : partitionRecords) {
                        String key = record.key();
                        String value = record.value();
                        long offset = record.offset();
                        long commitOffset = offset + 1; // 下一个消费的偏移量
                        System.out.printf("消费消息 key：%s, value：%s, 消息offset: %s, 提交offset: %s%n",
                                key, value, offset, commitOffset);

                        // 模拟消息处理过程
                        Thread.sleep(1500);
                    }

                    // 手动提交偏移量
                    consumer.commitSync();
                    System.out.println("已提交偏移量： " + partition);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 确保在关闭时提交偏移量
            consumer.close();
            System.out.println("消费者已关闭");
        }
    }
}
